<?php
namespace Plu\Library\Logger;

use Phalcon\Logger\Adapter,
    Phalcon\Logger\Exception,
    Phalcon\Logger\FormatterInterface,
    Phalcon\Logger\Formatter\Line as LineFormatter,
    Phalcon\Logger,
    Plu\Library\Services\KafkaService;

class Kafka extends Adapter {
    protected $name = "kafka";

    protected $options = [];

    protected $kafka;
    
    protected $logs = [];

    protected $isTransaction = false;

    public function __construct($name = "kafka", array $options = []) {
        if (!isset($options['topic'])) {
            throw new Exception("Parameter `topic` is required");
        }

        if (!isset($options['zookeeper'])) {
            throw new Exception("Parameter `zookeeper` is required");
        }

        if (!isset($options['path'])) {
            throw new Exception("Parameter `path` is required");
        }

        if (!empty($name)) {
            $this->name = $name;
        }

        $this->options = $options;

        register_shutdown_function([$this, 'commit']);
        register_shutdown_function([$this, 'close']);
    }

    public function getFormatter() {
        if (!$this->_formatter) {
            $this->_formatter = new LineFormatter();
        }
        return $this->_formatter;
    }

    /**
     * 写入日志
     */
    public function logInternal($message, $type, $time, $context = []) {
        switch ($type) {
            case Logger::SPECIAL:
                $loggerLevel = "SPECIAL";
                break;
            case Logger::CUSTOM:
                $loggerLevel = "CUSTOM";
                break;
            case Logger::DEBUG:
                $loggerLevel = "DEBUG";
                break;
            case Logger::INFO:
                $loggerLevel = "INFO";
                break;
            case Logger::NOTICE:
                $loggerLevel = "NOTICE";
                break;
            case Logger::WARNING:
                $loggerLevel = "WARNING";
                break;
            case Logger::ERROR:
                $loggerLevel = "ERROR";
                break;
            case Logger::ALERT:
                $loggerLevel = "ALERT";
                break;
            case Logger::CRITICAL:
                $loggerLevel = "CRITICAL";
                break;
            case Logger::EMERGENCE:
            case Logger::EMERGENCY:
                $loggerLevel = "EMERGENCY";
                break;
            default:
                $loggerLevel = "LOGGER";
                break;
        }

        $this->logs[] = [
            "date"     => date("Y-m-d H:i:s", $time),
            'time'     => microtime(true) * 1000,
            'type'     => $loggerLevel,
            'msg'      => $message,
            'context'  => $context,
            'hostname' => gethostname(),
            'pid'      => getmypid(),
            'name'     => $this->name
        ];

        if (!$this->isTransaction) {
            $this->send();
        }
    }

    /**
     * always return true
     */
    public function close() {
        return true;
    }

    public function begin() {
        $this->commit();
        $this->isTransaction = true;
    }

    public function commit() {
        if (!$this->isTransaction || empty($this->logs)) {
            $this->isTransaction = false;
            return;
        }

        $this->send();
        $this->isTransaction = false;
    }

    public function send() {
        if (empty($this->logs)) {
            return;
        }

        if ($this->kafka === null) {
            $this->kafka = new KafkaService($this->options['zookeeper'], $this->options['path'], $this->options['topic']);
        }

        while ($item = array_shift($this->logs)) {
            $this->kafka->produce($item);
        }

        $this->logs = [];
    }
}
